package juc.executors;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

/**
 * {@link CompletionService}
 * 只有一个实现类ExecutorCompletionService(Executor executor)，传入的是Executor
 * take()会block住，poll()不会block住
 * FutureTask是Future的默认实现
 * QueueingFuture#done() { completionQueue.add(task); }
 */
public class CompletionServiceExample2 {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        //testCompletionService();
        test();

    }

    private static void testCompletionService() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        final List<Callable<Integer>> callableList = Arrays.asList(
                () -> {
                    timeSleep(6);
                    System.out.println("the 1 finished");
                    return 1;
                }, () -> {
                    timeSleep(4);
                    System.out.println("the 2 finished");
                    return 2;
                });
        ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(executorService);
        List<Future<Integer>> futureList = new ArrayList<>();
        callableList.stream().forEach(callable -> {
            futureList.add(completionService.submit(callable));
        });

        /*Future<Integer> future = null;
        //take()是阻塞方法
        while ((future = completionService.take()) != null) {
            System.out.println(future.get());
        }
        */

        //poll()不是阻塞方法
        //Future<Integer> future = completionService.poll();
        Future<Integer> future = null;
        while ((future = completionService.poll(10, TimeUnit.SECONDS)) != null) {
            System.out.println(future.get());
        }
    }

    private static void test() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        ExecutorCompletionService<Event> completionService = new ExecutorCompletionService<>(executorService);
        final Event event = new Event(1);
        Future<Event> future = completionService.submit(new MyTask(event), event);
        //timeSleep(3);
        //System.out.println(event.result);
        System.out.println(completionService.take().get().result);
    }

    private static class MyTask implements Runnable{
        private final Event event;

        private MyTask(Event event) {
            this.event = event;
        }

        @Override
        public void run() {
            timeSleep(2);
            event.setResult("I am successfully");
        }
    }

    private static class Event {
        final private int eventId;
        private String result;

        private Event(int eventId) {
            this.eventId = eventId;
        }

        public int getEventId() {
            return eventId;
        }

        public String getResult() {
            return result;
        }

        public void setResult(String result) {
            this.result = result;
        }
    }

    /**
     * sleep the specify seconds
     */
    private static void timeSleep(long time) {
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
